package com.yujiahao.bigdata.sql

import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{DataFrame, Encoder, Encoders, SparkSession, functions}

object SparkSQL01_UDAF {
  def main(args: Array[String]): Unit = {
    //TODO 1、SparkSQL直接构建无法构建，包是私有的，使用伴生对象伴生类构建builder
    //SparkSQL对SparkCore进行了封装，所以环境也要发生变化

    val spark = SparkSession.builder
      .master("local")
      .appName("Word Count")
      .getOrCreate()
    //构建Dataframe
    val df: DataFrame = spark.read.json("data/user.json")

    //使用SQL的方式访问DataFrame

    df.createOrReplaceTempView("user")
    spark.sql("select name from user")

    //如果想在SQL语句中执行自定义的聚合函数，那么一般需要提前声明UDAF
    spark.udf.register("myAvg", functions.udaf(new MyAvgAge))
    spark.sql("select myAvg(age) from user").show()


    //TODO 2、关闭资源
    spark.stop()
  }

  //TODO 自定义函数。
  // 1、继承Aggregator类,注意要选择泛型是【IN ,BUFF,OUT 】。
  // 2、定义泛型--这里可以将缓冲里面的类型对象封装成一个对象进行调用
  // IN:输入数据的类型
  // BUFF:缓冲区的数据类型
  // OUT：聚合函数处理结果
  // 3、重写方法(2+4)
  class MyAvgAge extends Aggregator[Long, AvgVuffer, Long] {
    //缓冲区的初始化
    override def zero: AvgVuffer = {
      AvgVuffer(0L, 0L)
    }
    //将输入的年龄更新到缓冲区
    override def reduce(buff: AvgVuffer, age: Long): AvgVuffer = {
      buff.total += age
      buff.cnt += 1
      buff
    }
    //合并缓冲区
    override def merge(b1: AvgVuffer, b2: AvgVuffer): AvgVuffer = {
      b1.total += b2.total
      b1.cnt += b2.cnt
      b1
    }

    //计算完成
    override def finish(buff: AvgVuffer): Long = {

      buff.total / buff.cnt
    }

    //可以理解为序列化
    override def bufferEncoder: Encoder[AvgVuffer] = Encoders.product

    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }
//使用样例类将缓冲区里面的类型作为属性进行包装
  case class AvgVuffer(var total: Long, var cnt: Long)
}
